package com.lagou.rpc.consumer.client;

import com.lagou.rpc.consumer.config.RpcAddr;
import lombok.Data;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author liuc
 * @date 2021/6/7
 * @description
 **/
@Data
public class RpcClientForZk {

    private Map<String, RpcClientWrapper> rpcClientMap = new ConcurrentHashMap<>();
    private CuratorFramework zkClient;
    private String host;
    private int port;
    private String path;
    private RpcClientFactory factory;
/*
    public Object send(String msg) throws ExecutionException, InterruptedException {
        RpcClientWrapper client = searchClient();
        if(client == null){
            throw new NullPointerException("no rpc client available");
        }
        return client.send(msg);
    }

    private RpcClientWrapper searchClient(){
        //todo 根据最后一次请求时间, 或者随机获取
        Collection<RpcClientWrapper> values = rpcClientMap.values();
        if(CollectionUtils.isEmpty(values)){
            throw new NullPointerException("no rpc client available!");
        }
        RpcClientWrapper temp = null;
        long t = Integer.MAX_VALUE;
        for (RpcClientWrapper rcw : values) {
            if(rcw.getLastTime() == 0){
                //5秒内未被访问的
                return rcw;
            }
            if(t < rcw.getLastTime()){
                t = rcw.getLastTime();
                temp = rcw;
            }
        }
        return temp;
    }*/

    public void configZk(String host, int port, String path) throws Exception {
        if(zkClient == null){
            setConfig(host,port,path);
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
            zkClient = CuratorFrameworkFactory.newClient(host+":"+port, retryPolicy);
            zkClient.start();
            //先获取已有的
            List<NodeData> childrenData = getChildrenData();
            if(childrenData != null){
                //创建对应的rpcClient
                for (NodeData nd : childrenData) {
                    RpcAddr rpcAddr = new RpcAddr();
                    rpcAddr.setHost(nd.ip);
                    rpcAddr.setPort(nd.port);
                    rpcAddr.setLastTime(nd.lastTime);
                    factory.put(nd.path, rpcAddr);
                }
            }
            //监听节点
            watch();
        }
    }

    private void setConfig(String host, int port, String path){
        this.host = host;
        this.port = port;
        this.path = path;
    }

    public List<NodeData> getChildrenData() throws Exception {
        if(zkClient != null){
            List<String> list = zkClient.getChildren().forPath(path);
            if(list != null && !list.isEmpty()){
                List<NodeData> nodeDataList = new ArrayList<>();
                for (String s : list) {
                    NodeData nd = new NodeData(path+"/"+s,zkClient.getData().forPath(path+"/"+s));
                    nodeDataList.add(nd);
                }
                return nodeDataList;
            }
        }
        return null;
    }

    @Data
    class NodeData{
        private String path;
        private String data;
        private String ip;
        private int port;
        private long lastTime;

        public NodeData(){

        }

        public NodeData(String p, byte[] bytes){
            this.path = p;
            this.data = new String(bytes);
            String[] arr = this.data.split(";");
            String[] addr = arr[0].split(":");
            this.ip = addr[0];
            this.port = Integer.parseInt(addr[1]);
            if(arr.length > 1){
                lastTime = Long.parseLong(arr[1]);
            }
        }
    }

    private void watch(){
        CuratorCache curatorCache = CuratorCache.builder(zkClient, path).build();
        CuratorCacheListener listener = CuratorCacheListener.builder()
                .forCreates(node -> {
                    //System.out.printf("Node created: [%s]%n", node);
                    if(!node.getPath().equals(path)){
                        //子节点创建, 创建RpcClient
                        NodeData nd = new NodeData(node.getPath(),node.getData());
                        RpcAddr rpcAddr = new RpcAddr();
                        rpcAddr.setPort(nd.port);
                        rpcAddr.setHost(nd.ip);
                        rpcAddr.setLastTime(nd.lastTime);
                        factory.put(nd.path, rpcAddr);
                    }
                })
                .forChanges((oldNode, node) -> {
                    //System.out.printf("Node changed. Old: [%s] New: [%s]%n", oldNode, node);
                    NodeData nd = new NodeData(node.getPath(),node.getData());
                    System.out.println(nd);
                    RpcAddr rpcAddr = factory.get(nd.path);
                    if(rpcAddr == null){
                        rpcAddr = new RpcAddr();
                        rpcAddr.setHost(nd.ip);
                        rpcAddr.setPort(nd.port);
                        rpcAddr.setLastTime(nd.lastTime);
                        factory.put(nd.path, rpcAddr);
                    }else{
                        rpcAddr.setLastTime(nd.lastTime);
                    }
                })
                .forDeletes(oldNode -> {
                    //删除节点, 关闭RpcClient
                    //System.out.printf("Node deleted. Old value: [%s]%n", oldNode);
                    //NodeData nd = new NodeData(oldNode.getPath(),oldNode.getData());
                    factory.remove(oldNode.getPath());
                })
                .forInitialized(() -> System.out.println("Cache initialized"))
                .build();
        curatorCache.listenable().addListener(listener);
        curatorCache.start();
    }

}
